package com.hooper.iot.face.mq;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.hooper.iot.entity.DelFaceEntity;
import com.hooper.iot.cache.UnifyDeviceCache;
import com.hooper.iot.entity.addFace.AddFaceReq;
import com.hooper.iot.face.components.ChannelRepository;
import com.hooper.iot.face.queue.ConsumerQueueReposity;
import com.hooper.iot.resource.entity.DeviceFeignRes;
import com.hooper.iot.resource.entity.UnifyDevNotifyMessage;
import com.hooper.iot.resource.entity.req.CDevOperatePerson;
import com.hooper.iot.resource.entity.req.CGetIntegralPerson;
import com.hooper.iot.resource.entity.req.CmsImageInfoDto;
import com.hooper.iot.resource.entity.rsp.PersonInfoRspDto;
import com.hooper.iot.resource.impl.UnifyDevServiceImpl;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Slf4j
public class FaceKafkaListener {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    private UnifyDevServiceImpl unifyDevService;

    @Autowired
    private UnifyDeviceCache unifyDeviceCache;

    @Autowired
    private ChannelRepository channelRepository;

    @Value("${device.host}")
    private String baseUrl;

    @Autowired
    private ConsumerQueueReposity consumerQueueReposity;

    @Value("${device.maxSize}")
    private Integer maxSize;

    private static SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @KafkaListener(topics = "device-notify", groupId = "iot")
    private void deviceCache(ConsumerRecord record) {
        UnifyDevNotifyMessage message = JSONObject.parseObject((String) record.value(), UnifyDevNotifyMessage.class);
        unifyDevService.handleMessage(message);
    }

    @KafkaListener(topics = "usmp_syc_personinfo_nty", groupId = "iot")
    private void syncPersonInfoNty(ConsumerRecord record) {
        log.info("同步请求 ==> {}", record.value());
        PersonInfoRspDto rspDto = new PersonInfoRspDto();
        JSONObject message = JSONObject.parseObject((String) record.value());
        JSONArray devIdList = message.getJSONArray("devIdList");
        for (Object one : devIdList) {
            String devId = JSONObject.toJSONString(one);
            rspDto.setDevId(devId.replace("\"", ""));
            rspDto.setSycNum(20);
            DeviceFeignRes.DeviceData device = unifyDeviceCache.findDevice(rspDto.getDevId());
            if (device != null && channelRepository.get(device.getIpv4()) != null) {
                kafkaTemplate.send(TopicConstants.TOPIC_SYC_PERSONINFO_REQ, JSONObject.toJSONString(rspDto));
            } else {
                log.info("取消回应，设备离线");
            }
        }
    }

    @KafkaListener(topics = "usmp_syc_personinfo_rsp", groupId = "iot")
    private void syncPersonInfoRsp(ConsumerRecord record) throws ParseException {
        log.info("同步人员 ==> {}", record.value());
        CDevOperatePerson cDevOperatePerson = JSONObject.parseObject((String) record.value(), CDevOperatePerson.class);
        DeviceFeignRes.DeviceData device = unifyDeviceCache.findDevice(cDevOperatePerson.getDevId());
        log.info(JSONUtil.toJsonStr(device));
        log.info(JSONUtil.toJsonStr(channelRepository));
        if (device != null && device.getIpv4() != null && channelRepository.get(device.getIpv4()) != null) {
            Channel channel = channelRepository.get(device.getIpv4());
            Assert.notNull(channel, "设备未连接！");
            Map<String, List<String>> map = this.translateFaceData(cDevOperatePerson);
            if (map.get("add") != null && map.get("add").size() > 0) {
                for (String one : map.get("add")) {
                    log.info("新增人员 ==> {}", one);
                    channel.writeAndFlush(new TextWebSocketFrame(one));
                }
            }
            if (map.get("del") != null && map.get("del").size() > 0) {
                for (String one : map.get("del")) {
                    log.info("删除人员 ==> {}", one);
                    channel.writeAndFlush(new TextWebSocketFrame(one));
                }
            }
        } else {
            log.info(record.value() + "设备不在线！");
        }

    }

    public Map<String, List<String>> translateFaceData(CDevOperatePerson datas) throws ParseException {
        if (consumerQueueReposity.get(datas.getDevId())==null){
            consumerQueueReposity.put(datas.getDevId(), new AtomicInteger(1));
            log.info("==>设备：" + datas.getDevId() + "，已负载" + consumerQueueReposity.get(datas.getDevId()).get() + "条数据");
        }else {
            AtomicInteger integer = consumerQueueReposity.get(datas.getDevId());
            synchronized (integer) {
                int andAdd = consumerQueueReposity.get(datas.getDevId()).intValue();
                Assert.isTrue(andAdd < maxSize, datas.getDevId() + "设备超负载，暂不接收");
                int sum = consumerQueueReposity.get(datas.getDevId()).addAndGet(1);
                log.info("==>设备：" + datas.getDevId() + "，已负载" + sum + "条数据");
            }
        }
        Map<String, List<String>> map = new HashMap<>();
        List<String> dels = new ArrayList<>();
        List<String> adds = new ArrayList<>();
        if (datas != null && datas.getPesonInfoList().size() > 0) {
            AddFaceReq addFaceReq = new AddFaceReq();
            List<AddFaceReq.Body> bodyList = new ArrayList<>();

            DelFaceEntity delFaceEntity = new DelFaceEntity();
            List<DelFaceEntity.Body> delBodyList = new ArrayList<>();

            for (CGetIntegralPerson one : datas.getPesonInfoList()) {
                if (one.getAction() != 2) {
                    AddFaceReq.Body body = addFaceReq.new Body();
                    List<AddFaceReq.Images> imagesList = new ArrayList<>();
                    for (CmsImageInfoDto image : one.getImageList()) {
                        AddFaceReq.Images images = addFaceReq.new Images();
                        images.setData(baseUrl + image.getImageData());
                        images.setFileFormat(image.getImageFormat());
                        images.setImageType("Face");
                        imagesList.add(images);
                    }
                    body.setControltype(one.getControltype());
                    //TODO 指纹和卡暂不适配
                    body.setAccessCards(null);
                    body.setFacialFeature(null);
                    body.setFingerprint(null);
                    if (CollectionUtil.isNotEmpty(one.getIdentityList())) {
                        body.setIdentityNo(one.getIdentityList().get(0).getIdentityNo());
                        body.setIdentityType(getIdentityType(one.getIdentityList().get(0).getIdentityType()));
                    }
                    body.setMatchType("Face");
                    body.setName(one.getName());
                    //TODO 字段匹配
                    body.setPersonAuthType(authType(Integer.getInteger(one.getPersonAuthType())));
                    body.setPersonType("AVERAGE");
                    if (!StringUtils.isEmpty(one.getPersonExpiryDate())) {
                        body.setPersonExpiryDate(sf.parse(one.getPersonExpiryDate()).getTime());
                    }
                    body.setPersonId(one.getPersonId());
                    body.setSex(Integer.getInteger(one.getSex()));
                    body.setImages(imagesList);
                    bodyList.add(body);

                } else {
                    DelFaceEntity.Body body = delFaceEntity.new Body();
                    body.setPersonId(one.getPersonId());
                    body.setIdentityNo(one.getIdentityList().get(0).getIdentityNo());
                    delBodyList.add(body);
                }
                //新增目前单条下发，多条下发把这块放外面就可以了
                addFaceReq.setBody(bodyList);
                addFaceReq.setCallBack(true);
                addFaceReq.setDeviceId(datas.getDevId());
                addFaceReq.setMsgId(one.getUptKey());
                addFaceReq.setMsgType("SEND_FACE_ID");
                if (bodyList.size() > 0) {
                    adds.add(JSONObject.toJSONString(addFaceReq));
                }
                //删除
                delFaceEntity.setBody(delBodyList);
                delFaceEntity.setCallBack(true);
                delFaceEntity.setDeviceId(datas.getDevId());
                delFaceEntity.setMsgId(one.getUptKey());
                delFaceEntity.setMsgType("DEL_FACE_ID");
                if (delBodyList.size() > 0) {
                    dels.add(JSONObject.toJSONString(delFaceEntity));
                }
            }
            map.put("add", adds);
            map.put("del", dels);
        }
        return map;
    }

    private String getIdentityType(Integer identityType) {
        if (identityType == 1) {
            return "IDCARD";
        } else if (identityType == 2) {
            return "SOCIALCARD";
        } else if (identityType == 3) {
            return "PASSPORT";
        } else if (identityType == 4) {
            return "EMPLOYEE";
        } else {
            return "OTHER";
        }
    }

    private String authType(Integer type) {
        if (type != null && type == 1) {
            return "VISITOR";
        } else {
            return "ADMIN";
        }
    }
}
